package com.demo.spark.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.ForeachFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.rdd.RDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.collection.Seq;

import java.util.Arrays;
import java.util.List;

public class DataFrameTemplate {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setMaster("local").setAppName("data frame ");
        JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);
        sparkContext.setLogLevel("error");
        SQLContext sqlContext = new SQLContext(sparkContext);
        List<Integer> integers = Arrays.asList(1, 2, 3, 4);
        JavaRDD<Integer> parallelize = sparkContext.parallelize(integers,2);
        JavaRDD<Person> persons = parallelize.map(new Function<Integer, Person>() {
            @Override
            public Person call(Integer v1) throws Exception {
                Person p = new Person();
                p.setName("zhangsan");
                p.setAge(21);
                p.setSalary(2200);
                p.setId(v1);
                return p;
            }
        });
        Dataset<Row> dataset = sqlContext.createDataFrame(persons, Person.class);
        dataset.createOrReplaceTempView("person");
        String[] columns = dataset.columns();
        for (String column : columns) {
            System.out.println(column);
        }
        long count = dataset.count();
        System.out.println("count : "+count);
        String[] strings = sqlContext.tableNames();
        for (String string : strings) {
            System.out.println(string);
        }
        Dataset<Row> sql = sqlContext.sql("select * from Person");
        long count1 = sql.count();
        System.out.println("sql count : " + count1);
        sql.foreach(new ForeachFunction<Row>() {
            @Override
            public void call(Row row) throws Exception {
                System.out.println(row.toString());
            }
        });
    }

}
